package cn.itcast.streaming.map;

import cn.itcast.streaming.entity.ItcastDataPartObj;
import cn.itcast.streaming.entity.VehicleLocationModel;
import cn.itcast.streaming.utils.GeoHashUtil;
import cn.itcast.streaming.utils.RedisUtil;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * Author itcast
 * Date 2021/9/27 10:29
 * 将实时上报的车辆的数据的经纬度位置信息转换成带有国家、省市区、地理位置的数据拉宽
 * 从redis中读取车辆的位置数据，根据geohash来获取
 * 开发步骤：
 * 1.实现open方法连接 redis
 * 2.实现map 方法，用于通过geohash来读取redis中地理数据并转换成对象拉宽 ItcastDataPartObj
 * 3.收集数据
 * 4.关闭连接
 */
public class LocationInfoRedisFunction extends RichMapFunction<ItcastDataPartObj, ItcastDataPartObj> {
    @Override
    public ItcastDataPartObj map(ItcastDataPartObj value) throws Exception {
        //1.1.获取车辆数据的经度和维度生成 geohash (经度,维度)-> geohash 字符串 -> 地理详细位置
        Double lng = value.getLng();
        Double lat = value.getLat();
        String geohash = GeoHashUtil.encode(lat, lng);
        //1.2.根据geohash 从redis中获取value值（geohash在redis中是作为主键存在）
        byte[] locationDetailArr = RedisUtil.get(Bytes.toBytes(geohash));
        //1.3.如果查询出来的值不为空，将其通过JSON对象转换成 VehicleLocationModel 对象，否则置为 null
        if (locationDetailArr != null && locationDetailArr.length > 0) {
            String vehicleLocationJson = Bytes.toString(locationDetailArr);
            VehicleLocationModel model = JSON
                    .parseObject(vehicleLocationJson, VehicleLocationModel.class);
            //1.4.如果当前对象不为空，将国家，省市区地址赋值给 itcastDataPartObj，否则置为 null
            if (model != null) {
                value.setCountry(model.getCountry());
                value.setProvince(model.getProvince());
                value.setCity(model.getCity());
                value.setDistrict(model.getDistrict());
                value.setAddress(model.getAddress());
            } else {
                value.setCountry(null);
                value.setProvince(null);
                value.setCity(null);
                value.setDistrict(null);
                value.setAddress(null);
            }
        } else {
            value.setCountry(null);
            value.setProvince(null);
            value.setCity(null);
            value.setDistrict(null);
            value.setAddress(null);
        }

        //1.5.返回数据
        return value;
    }
}
